package middleware

import (
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
	"os"
	"os/signal"
	"syscall"
	"time"
)

type XCMessage struct {
}

var producer *kafka.Producer
var deliveryChan = make(chan kafka.Event)

var bootstrapServers = "192.168.56.130:9092"
var topic = "request-topic"

//var sysLogTopic = "logs-system"

type SendMessage interface {
	Send(context string)
}
type ConsumerMessage interface {
	Accept(topic string, msgChan <-chan string)
}

var run = true

func init() {

	var err error
	producer, err = kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": bootstrapServers})

	if err != nil {
		logger.Info(fmt.Sprintf("Failed to create producer: %v", err))
		os.Exit(1)
	}
	listenSendMessageStatus()

	go func() {
		// 等待中断信号
		quit := make(chan os.Signal, 1)
		signal.Notify(quit, os.Interrupt)
		<-quit
		run = false
		logger.Info("Shutting down server...")
	}()

}
func (xcMessage XCMessage) Send(context string) {
	var err = producer.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value:          []byte(context),
	}, deliveryChan)

	if err != nil {
		if err.(kafka.Error).Code() == kafka.ErrQueueFull {
			time.Sleep(time.Second)
		}
		logger.ERROR(fmt.Sprintf("Failed to produce message: %v", err))
	}
}

func listenSendMessageStatus() {

	go func() {
		// 设置信号处理，优雅退出
		sigchan := make(chan os.Signal, 1)
		signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

		// 等待消息传递报告或中断信号
		for {
			select {
			case <-sigchan:
				logger.Info("Interrupt received, shutting down...")
				return
			case e := <-deliveryChan:
				// 消息传递事件
				m := e.(*kafka.Message)
				if m.TopicPartition.Error != nil {
					logger.ERROR(fmt.Sprintf("Delivery failed: %v", m.TopicPartition.Error))
				}
				return
			}
		}
	}()
}

func (xcMessage XCMessage) Accept(topic string, msgChan chan<- string) {
	c, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": bootstrapServers,
		"group.id":          "myGroup",
		"auto.offset.reset": "earliest",
	})

	if err != nil {
		logger.ERROR(fmt.Sprintf("NewConsumer error %v", err))
	}

	err = c.SubscribeTopics([]string{topic}, nil)

	if err != nil {
		logger.ERROR(fmt.Sprintf("SubscribeTopics error %v", err))
	}

	go func() {
		for run {
			msg, err := c.ReadMessage(time.Second)
			if err == nil {
				msgChan <- string(msg.Value)
				//logger.Info(fmt.Sprintf("consumer message on %s: %s", msg.TopicPartition, string(msg.Value)))
				//c.Commit()
			} else if !err.(kafka.Error).IsTimeout() {
				logger.ERROR(fmt.Sprintf("consumer error: %v (%v)", err, msg))
			}
		}
		c.Close()
	}()

}
